package delivery

import (
	"fmt"
	"learning/gooop/saga/mqs/eventbus"
	"learning/gooop/saga/mqs/logger"
	"learning/gooop/saga/mqs/models"
	"learning/gooop/saga/mqs/models/events"
	"time"
)

type tLiveMsgSource struct {
	clientID string
	expireFunc tSourceExpireFunc
	msgChan chan *models.QueuedMsg
}

func newLiveMsgSource(clientID string, expireFunc tSourceExpireFunc) iMsgSource {
	it := new(tLiveMsgSource)
	it.init(clientID, expireFunc)
	return it
}

func (me *tLiveMsgSource) init(clientID string, expireFunc tSourceExpireFunc) {
	me.clientID = clientID
	me.expireFunc = expireFunc
	me.msgChan = make(chan *models.QueuedMsg, gMaxLiveMsgs)

	eventbus.GlobalEventBus.Sub(events.MsgPublishedEvent,
		me.id(),
		me.handleMsgPublished)
	go me.beginWatchExpire()
}


func (me *tLiveMsgSource) id() string {
	return fmt.Sprintf("tLiveMsgSource.%s", me.clientID)
}

func (me *tLiveMsgSource) beginWatchExpire() {
	for range time.Tick(1 * time.Second) {
		if me.expireFunc() {
			me.afterExpired()
			return
		}
	}
}

func (me *tLiveMsgSource) afterExpired() {
	eventbus.GlobalEventBus.Unsub(events.MsgPublishedEvent, me.id())
	close(me.msgChan)
}

func (me *tLiveMsgSource) handleMsgPublished(_ string, args interface{}) {
	msg, ok := args.(*models.QueuedMsg)
	if !ok {
		return
	}

	if msg.ClientID != me.clientID {
		return
	}

	if len(me.msgChan) > gMaxLiveMsgs {
		return
	}

	logger.Logf(
		"tLiveMsgSource.handleMsgPublished, clientID=%s, msg=%s/%s/%s",
		me.clientID, msg.GlobalID, msg.SubID, msg.Topic )
	me.msgChan <- msg
}

func (me *tLiveMsgSource) MsgChan() <- chan *models.QueuedMsg {
	return me.msgChan
}

var gMaxLiveMsgs = 100